package cn.kimming.rpc.remoting.server;

import cn.kimming.rpc.remoting.codec.MyRpcProtocolServerCodec;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class NettyServer extends AbstractNetworkServer {

	private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);

	private NioEventLoopGroup bossGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors());
	private NioEventLoopGroup workerGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2);
	private ServerBootstrap serverBootstrap;

	private ExecutorService bussinessPool = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors() * 2,
			Runtime.getRuntime().availableProcessors() * 48,
			60,
			TimeUnit.SECONDS,
			new SynchronousQueue<>(),
			new MyRpcThreadFactory(),
			new ThreadPoolExecutor.CallerRunsPolicy());	// 业务线程池

	@Override
	protected void start(int port, ServiceMap serviceMap) {
		serverBootstrap = new ServerBootstrap()
				.group(bossGroup, workerGroup)
				.channel(NioServerSocketChannel.class)
				.option(ChannelOption.SO_REUSEADDR, true)
				.childOption(ChannelOption.SO_KEEPALIVE, true)
				.childOption(ChannelOption.TCP_NODELAY, true)
				.childHandler(new ChannelInitializer<NioSocketChannel>() {
					@Override
					protected void initChannel(NioSocketChannel ch) throws Exception {
						ch.pipeline()
								.addLast(new MyRpcProtocolServerCodec())
								.addLast(new ServerHandler(serviceMap, bussinessPool));
					}
				});

		try {
			ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
			logger.info("[my-rpc] rpc server started successfully and listen on port " + port);
			channelFuture.channel().closeFuture().sync();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			bossGroup.shutdownGracefully();
			workerGroup.shutdownGracefully();
		}
	}
}
